#include <pthread.h>
#include <fcntl.h>
#include <mqueue.h>
#include "server.h"
#include "ssl.h"
#include "db.h"
#include "ecmfs_cmn.h"
#include <sys/syscall.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>


ecmfs_msg_queue_t *master_mq;
   
#define DEBUG 0

#if DEBUG
#define DEBUG_LOG(fmt , args...) \
    printf("\n[%s][%d] " fmt "\n" , __FUNCTION__, __LINE__ , ##args );
#else
#define DEBUG_LOG(fmt , args... ) 
#endif

int create_master_thread(void)
{
    int retval;
    int err_num;
    server_thread_info_t  *tinfo;

    master_mq = ecmfs_message_queue_init();

    if (master_mq == NULL) {
    	fprintf(stderr,
            "Failed to create msgq for master thread.");
    	return ECMFS_FALSE;
    }

   tinfo = malloc(sizeof(server_thread_info_t));
   if (!tinfo) {
         DEBUG_LOG(
             "Failed to alloc for edge thread");
		ecmfs_message_queue_deinit(master_mq);         
         return ECMFS_FALSE;
   }

	retval = pthread_create( &tinfo->thread_id,
		     NULL,
		     master_thread,
		     (void*)tinfo);
	
	if (retval < 0 ) {
		err_num = errno;
		DEBUG_LOG(
	        "Failed to create master thread. errno=%d [%s]", err_num, 
			strerror(err_num));
		ecmfs_message_queue_deinit(master_mq);
	    free(tinfo);
	    retval = ECMFS_FALSE;
	} else {
	   DEBUG_LOG(
	    "Created master thread. thr_idx = %d", tinfo->tid);
	   retval = ECMFS_TRUE;
	}
	
    return(retval);
        
} 

void master_handle_message(unsigned char *buffer, int bufsize, int *more)  
{
    unsigned char *decoded_buffer; 
    ecmfs_message_t *msg;
    int decoded_bufsize;
    
    if (bufsize < sizeof(ecmfs_message_t)) {
        fprintf (stderr, "Core: cannot handle this message\n");
        return;
    }
        
    msg = (ecmfs_message_t *) buffer;
    if (msg->magic_num == 19977991) {
      /* Can get clear text for initial
         user add. A TODO is to make that
         an SSL connection, where the master
         does not know the user's public key
         yet.
       */
      fprintf (stderr, "Server: got cleartext\n");
    } else {
      /* Check the socket properties
         client sends data encrypted 
         in symmetric key
         Other nodes encrypt using
         self public key 
       */      
      fprintf (stderr, "Server: got non ecmfs header check if encrypted\n");
      decode_rsa_message(&ecmfs_ssl, ecmfs_ssl.enc_priv_key, buffer, bufsize,
                    &decoded_buffer, &decoded_bufsize); 
      msg = (ecmfs_message_t *) decoded_buffer;
    }
    fprintf(stderr, "Message from:");      
    switch(msg->from) 
    {
     case NODE_CLIENT:
         fprintf(stderr, "Client : \n");         
         master_handle_client_message(msg, more);
         break;
     case NODE_EDGE:
         fprintf(stderr, "Edge : \n");         
         master_handle_edge_message(msg, more);
         break;
     case NODE_CORE:
         fprintf(stderr, "Core : \n");         
         master_handle_core_message(msg, more);
         break;
     case NODE_MASTER:
         fprintf(stderr, "Master : \n");         
         master_handle_master_message(msg, more);
         break;
     default:
         fprintf(stderr, "Unknown : \n");         
         break;
    }
}

void *
master_thread (void *arg)
{
    pid_t tid, pid;
    server_thread_info_t *thread_info;
    unsigned char *buffer;   
    int bytes_read;
    int more = 0;
    int sd; 
    int ret;
    pthread_detach (pthread_self());

    if (master_mq == NULL) {
	    fprintf(stderr,"No queue to wait in %s. Exiting",
                     __FUNCTION__);
        pthread_exit(NULL);
    }
    

    if (!arg) {
    	fprintf(stderr,"NULL arg to %s. Exiting", __FUNCTION__);
	    pthread_exit(NULL);
    }

    //master_signal_blocker();
    
    thread_info = (server_thread_info_t *)arg;

    tid = (pid_t)syscall(SYS_gettid);
    pid = getpid();

    thread_info->pid = pid;
    thread_info->tid = tid;
    
    while(1) {
        
        /* timed block */
        while (1) {
        if (ecmfs_message_queue_count(master_mq) <= 0) {
            usleep(200);
        } else {
            break;
        }
        }
        
        ret = ecmfs_message_queue_del(master_mq,
                    (char **) &buffer, &bytes_read, &sd);
                    
        if ((ret == -1) ||(bytes_read == 0))
        {
           //fprintf(stderr,"No message");
           continue;
        } else {
            //handle the msg
            more = sd;
            master_handle_message(buffer,bytes_read, &more);
            free(buffer);
        }
    }
    pthread_exit(NULL);
}

bool_ecmfs  master_msg_send (unsigned char *buffer, int buff_size, int sd)
{
    int ret;
    ecmfs_message_t *msg;

    DEBUG_LOG("Send %p with size %d\n", buffer, buff_size);
   
    if (master_mq == NULL) {
        fprintf(stderr,"\n No queue handle found :%s",
            
                __FUNCTION__);
        return ECMFS_FALSE;
    }

    msg = (ecmfs_message_t *) buffer;
    DEBUG_LOG("Enqueing msg %p, msg->to=%d, type %d accept sd=%d\n", 
                msg, msg->to, msg->type, sd);
    dump_ecmfs_message(msg);

    ret = ecmfs_message_queue_add (master_mq, (void *)buffer, 
                         buff_size,sd);
    if (ret != 0) {
        fprintf(stderr,"\n FRR : failed to process_enqueue :%s",
                __FUNCTION__);
        return ECMFS_FALSE;
    }

    return ECMFS_TRUE;
}


int store_user_public_key(ecmfs_add_user_t *addMsg);
bool_ecmfs  master_handle_client_message(ecmfs_message_t *msg, int *more)
{
    int accept_sd;
    
    accept_sd = *more;
    
    switch(msg->type) {
    case CLIENT_OP_HELLO:
        {
            db_ret_t db_ret;
            bool_ecmfs userAdded = ECMFS_FALSE;
            ecmfs_add_user_t *addMsg;
            
            addMsg = (ecmfs_add_user_t *) msg->message;
            DEBUG_LOG("request to check user %s\n", addMsg->user_id);

        
            db_ret = open_db();

            if (db_ret != DB_SUCCESS) {
                fprintf(stderr, 
                "Error in opening db %d", 
                db_ret);
            }
        
        
            db_ret = check_user((char *)addMsg->user_id);
            if (db_ret == DB_SUCCESS) {
                userAdded = ECMFS_TRUE;
            }

            if (userAdded == ECMFS_FALSE) {
                DEBUG_LOG("user %s not present: send who are you\n", addMsg->user_id);
                send_response(accept_sd, addMsg->user_id,
                    SERVER_RESP_WHOAREYOU, ECMFS_FALSE, NODE_CLIENT);    
            } else {
                DEBUG_LOG("user %s present:send hello\n", addMsg->user_id);
                send_response(accept_sd, addMsg->user_id,
                    SERVER_RESP_SENDING_PUBKEY, ECMFS_FALSE, NODE_CLIENT);    
            }
        }
        break;
    case CLIENT_OP_ADDUSER:
        {
            db_ret_t db_ret;
            bool_ecmfs userAdded = ECMFS_FALSE;
            
            ecmfs_add_user_t *addMsg;
            addMsg = (ecmfs_add_user_t *) msg->message;


        
            db_ret = open_db();

            if (db_ret != DB_SUCCESS) {
                fprintf(stderr, 
                "Error in opening db %d", 
                db_ret);
                send_response(accept_sd, addMsg->user_id,
                    SERVER_RESP_ERROR, ECMFS_FALSE,NODE_CLIENT);                 
            } else {
                db_ret = check_user((char *)addMsg->user_id);
                if (db_ret == DB_SUCCESS) {
                    userAdded = ECMFS_TRUE;
                    store_user_public_key(addMsg);
                    send_response(accept_sd, addMsg->user_id,
                    SERVER_RESP_SENDING_PUBKEY, ECMFS_FALSE, NODE_CLIENT);                    
                }

                if (userAdded == ECMFS_FALSE) {
                    DEBUG_LOG( 
                    "Add to %s to DB\n",addMsg->user_id );
                    db_ret = add_user((char *)addMsg->user_id,
                                    NULL);
                    if (db_ret == DB_SUCCESS) {
                        db_ret = close_db();
                        store_user_public_key(addMsg);
                        send_response(accept_sd, addMsg->user_id,
                        SERVER_RESP_SENDING_PUBKEY, ECMFS_FALSE, NODE_CLIENT);                        
                    } else {
                        db_ret = close_db();
                        send_response(accept_sd, addMsg->user_id,
                        SERVER_RESP_ERROR, ECMFS_FALSE, NODE_CLIENT);                        
                    }
                }
            }
        }
        break;
    case CLIENT_OP_REMUSER:
        {
            db_ret_t db_ret;
            bool_ecmfs userAdded = ECMFS_TRUE;
            
            ecmfs_add_user_t *addMsg;
            addMsg = (ecmfs_add_user_t *) msg->message;
            printf("request to check user %s", addMsg->user_id);
            printf("public key is %s", addMsg->user_id);

        
            db_ret = open_db();

            if (db_ret != DB_SUCCESS) {
                fprintf(stderr, 
                "Error in opening db %d", 
                db_ret);
                send_response(accept_sd, addMsg->user_id,
                SERVER_RESP_ERROR, ECMFS_TRUE, NODE_CLIENT);                
            } else {
                db_ret = check_user((char *)addMsg->user_id);
                if (db_ret != DB_SUCCESS) {
                    userAdded = ECMFS_FALSE;
                    send_response(accept_sd, addMsg->user_id,
                    SERVER_RESP_DONE, ECMFS_TRUE, NODE_CLIENT); 
                }

                if (userAdded == ECMFS_TRUE) {
                    DEBUG_LOG( 
                    "REMOVE %s from DB\n",addMsg->user_id );
                    /* Close  existing sessions of user */
                    /* Remove the name from filegroups  */
                    /* Associate files with owner as 
                       this user to root  */
                    /* Finally remove the user */
                    db_ret = del_user((char *)addMsg->user_id);
                    if (db_ret == DB_SUCCESS) {
                        db_ret = close_db();
                        send_response(accept_sd, 
                            addMsg->user_id,
                        SERVER_RESP_DONE, ECMFS_TRUE, NODE_CLIENT);                  
                    } else {
                        db_ret = close_db();
                        send_response(accept_sd, 
                            addMsg->user_id,
                        SERVER_RESP_ERROR, ECMFS_TRUE,NODE_CLIENT);  
                    }
                }
            }
        }
        break;
    case CLIENT_OP_MOUNT:
        {
            db_ret_t db_ret;
            bool_ecmfs userAdded = ECMFS_FALSE;
            
            ecmfs_add_user_t *addMsg;
            addMsg = (ecmfs_add_user_t *) msg->message;
        
            db_ret = open_db();

            if (db_ret != DB_SUCCESS) {
                fprintf(stderr, 
                "Error in opening db %d", 
                db_ret);
                send_response(accept_sd, addMsg->user_id,
                    SERVER_RESP_ERROR, ECMFS_FALSE,NODE_CLIENT);                 
            } else {
                db_ret = check_user((char *)addMsg->user_id);
                if (db_ret == DB_SUCCESS) {
                    userAdded = ECMFS_TRUE;
                }

                if (userAdded == ECMFS_TRUE) {
                    session_aes_key_t *sess;
                    unsigned char *key, *iv;
                    node_details_t *nodes;
                    int num_nodes;
                    int i;
                    int clientRespLen;
                    
                    clientRespLen = sizeof(session_aes_key_t);
                                    
                    sess = (session_aes_key_t *) malloc(clientRespLen);
                    if (NULL == sess) {
                        /* ignore nothing to do here
                            client has to timeout 
                         */
                        return ECMFS_FALSE;     
                        
                    } 
                    
                    memset(sess, 0, clientRespLen);
                    
                    /* generate a session key */
                    ssl_gen_aes_keys(&ecmfs_ssl, &key, &iv);
                    /* add session key and edge user 
                       to response structure */
                    get_nodes(&nodes, &num_nodes);
                    
                    memcpy(sess->key, key, AES_KEYLEN/8);
                    memcpy(sess->iv,  iv,  AES_KEYLEN/8);
                    sess->edge_port = 0;
                    for(i = 0; i < num_nodes; i++)
                    {
                        if (strncmp(nodes->node_name,"edge", 4) == 0)
                        {
                            memcpy(sess->edge_name,  nodes->node_name,  32);
                            memcpy(sess->edge_ip,  nodes->node_ip,  32);
                            sess->edge_port = nodes->port;
                            break;
                        }
                    }
                    
                    memcpy(sess->user_id,  addMsg->user_id,  32);
                    //memcpy(sess->user_ip,  addMsg->user_ip,  32);
                    //sess->edge_port = nodes->node_port;
                    
                    if ( sess->edge_port == 0 ) {
                        send_response(accept_sd, 
                            addMsg->user_id,
                        SERVER_RESP_ERROR, ECMFS_FALSE,NODE_CLIENT); 
                    } else {
                        /* send success response to client */
                        send_response(accept_sd, 
                            sess,
                        SERVER_RESP_MOUNT_OK, ECMFS_FALSE, NODE_CLIENT);                        
                    }

                    /* notify session key and user 
                       info to edge node also */
                    send_notify(sess->edge_ip,sess->edge_port, sess,
                           SERVER_NOTIFY_SESSION_UPDATE, ECMFS_FALSE, NODE_EDGE);
                    
                    /* send notify to edge server */
                    free(key);
                    free(iv);
                    free(nodes);
                    free(sess);
                }
            }
        }
        break;
    case CLIENT_OP_UNMOUNT:
        {
        }        
    }

    return ECMFS_TRUE;
}

bool_ecmfs  master_handle_edge_message(ecmfs_message_t *msg, int *more)
{
    int accept_sd;
    accept_sd = *more;
    switch(msg->type) {
    case SERVER_OP_HELLO:
        {
            db_ret_t db_ret;
            bool_ecmfs nodeAdded = ECMFS_FALSE;
            ecmfs_runtime_state_t *addMsg;
            
            addMsg = (ecmfs_runtime_state_t *) msg->message;
            DEBUG_LOG("request to check node %s\n", addMsg->info.node_name);

        
            db_ret = open_db();

            if (db_ret != DB_SUCCESS) {
                fprintf(stderr, 
                "Error in opening db %d", 
                db_ret);
            }
        
        
            db_ret = check_node((char *)addMsg->info.node_name);
            if (db_ret == DB_SUCCESS) {
                nodeAdded = ECMFS_TRUE;
            }
            
            if (nodeAdded == ECMFS_FALSE) {
                DEBUG_LOG("node %s not present: adding\n",
                    addMsg->info.node_name);
                
                DEBUG_LOG( 
                "Add to %s to DB\n",addMsg->info.node_name );
                db_ret = add_node(&addMsg->info);
                if (db_ret == DB_SUCCESS) {
                    db_ret = close_db();
                    store_node_public_key(addMsg);
                    send_response(accept_sd, addMsg->info.node_name,
                    SERVER_RESP_SENDING_PUBKEY, ECMFS_TRUE, NODE_EDGE); 
                    set_server_state(SERVER_STATE_EDGE_READY);
                } else {
                    db_ret = close_db();
                    send_response(accept_sd, addMsg->info.node_name,
                    SERVER_RESP_ERROR, ECMFS_TRUE, NODE_EDGE);                        
                }
            } else {
                store_node_public_key(addMsg);
                DEBUG_LOG("node %s already present:send hello\n", addMsg->info.node_name);
                send_response(accept_sd, addMsg->info.node_name,
                    SERVER_RESP_HELLO, ECMFS_FALSE, NODE_EDGE); 
                set_server_state(SERVER_STATE_EDGE_READY);
    /*                    
   
    set_server_state(SERVER_STATE_READY);
    set_server_state(SERVER_STATE_CORE_READY);
    set_server_state(SERVER_STATE_INTERNAL_ERROR);
    */                
            }
        }
        break;
    default:
        break;
    }

    return ECMFS_TRUE;
}



bool_ecmfs  master_handle_core_message(ecmfs_message_t *msg, int *more)
{
    int accept_sd;
    accept_sd = *more;
    switch(msg->type) {
    case SERVER_OP_HELLO:
        {
            db_ret_t db_ret;
            bool_ecmfs nodeAdded = ECMFS_FALSE;
            ecmfs_runtime_state_t *addMsg;
            
            addMsg = (ecmfs_runtime_state_t *) msg->message;
            DEBUG_LOG("request to check node %s\n", addMsg->info.node_name);

        
            db_ret = open_db();

            if (db_ret != DB_SUCCESS) {
                fprintf(stderr, 
                "Error in opening db %d", 
                db_ret);
            }
        
        
            db_ret = check_node((char *)addMsg->info.node_name);
            if (db_ret == DB_SUCCESS) {
                nodeAdded = ECMFS_TRUE;
            }

            if (nodeAdded == ECMFS_FALSE) {
                DEBUG_LOG("node %s not present: adding\n",
                    addMsg->info.node_name);
                
                DEBUG_LOG( 
                "Add to %s to DB\n",addMsg->info.node_name );
                db_ret = add_node(&addMsg->info);
                if (db_ret == DB_SUCCESS) {
                    db_ret = close_db();
                    store_node_public_key(addMsg);
                    send_response(accept_sd, addMsg->info.node_name,
                    SERVER_RESP_SENDING_PUBKEY, ECMFS_TRUE, NODE_CORE);
                    set_server_state(SERVER_STATE_CORE_READY);                    
                } else {
                    db_ret = close_db();
                    
                    send_response(accept_sd, addMsg->info.node_name,
                    SERVER_RESP_ERROR, ECMFS_TRUE, NODE_CORE);                        
                }
            } else {
                DEBUG_LOG("node %s already present:send hello\n", addMsg->info.node_name);
                store_node_public_key(addMsg);
                send_response(accept_sd, addMsg->info.node_name,
                    SERVER_RESP_HELLO, ECMFS_FALSE, NODE_CORE); 
                    set_server_state(SERVER_STATE_CORE_READY);
            }
        }
        break;
    default:
        break;
    }


    return ECMFS_TRUE;
}
bool_ecmfs  master_handle_master_message(ecmfs_message_t *msg, int *more)
{
    dump_ecmfs_message(msg);
    return ECMFS_TRUE;
}
